package com.ssm.subject.infra.basic.es;

import com.alibaba.fastjson.JSON;
import lombok.extern.log4j.Log4j2;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.math.NumberUtils;
import org.apache.http.HttpHost;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.TermQueryBuilder;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.elasticsearch.search.Scroll;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
import org.elasticsearch.search.sort.ScoreSortBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.io.IOException;
import java.util.*;

/**
 * es自定义工具类
 */
@Component
@Log4j2
public class EsRestClient {
    //配置es集群统一连接管理
    /**
     * es集群集合  key：集群名称 value：多节点信息(ip+端口)
     */
    private static Map<String, RestHighLevelClient> clientMap = new HashMap<>();

    @Resource
    private EsConfigProperties esConfigProperties;

    //发送给 Elasticsearch 请求时的通用配置选项
    private static final RequestOptions COMMON_OPTIONS;

    static { //使用 static 块进行初始化，确保在类加载时只执行一次
        RequestOptions.Builder builder = RequestOptions.DEFAULT.toBuilder();
        COMMON_OPTIONS = builder.build();
    }

    /**
     * 初始化集群连接
     * @PostConstruct: spring初始化bean后执行
     */
    @PostConstruct
    public void initialize() {
        List<EsClusterConfig> esConfigs = esConfigProperties.getEsConfigs();
        for(EsClusterConfig esClusterConfigfig : esConfigs) {
            log.info("initialize.config.name:{}, node:{}", esClusterConfigfig.getName(), esClusterConfigfig.getNodes());
            RestHighLevelClient restHighLevelClient = initRestClient(esClusterConfigfig);
            if(restHighLevelClient != null) { //集群连接成功
                clientMap.put(esClusterConfigfig.getName(), restHighLevelClient);
            } else {
                log.error("config.name:{}, node:{}.initError", esClusterConfigfig.getName(), esClusterConfigfig.getNodes());
            }
        }
    }

    /**
     * 初始化es Map集群集合的value RestHighLevelClient信息
     */
    public RestHighLevelClient initRestClient(EsClusterConfig esClusterConfig) {
        //分割出当前集群下的多个节点
        String[] ipPortArr = esClusterConfig.getNodes().split(",");
        //HttpHost集合可存放ip，端口信息等
        List<HttpHost> httpHostList = new ArrayList<>();
        //遍历把每个节点拆为ip和端口，存入HttpHost集合
        for(String ipPort : ipPortArr) {
            String[] ipPortInfo = ipPort.split(":");
            if(ipPortInfo.length == 2) {
                HttpHost httpHost = new HttpHost(ipPortInfo[0], NumberUtils.toInt(ipPortInfo[1]));
                httpHostList.add(httpHost);
            }
        }
        //把HttpHost集合转为数组
        HttpHost httpHosts[] = new HttpHost[httpHostList.size()];
        httpHostList.toArray(httpHosts);

        RestClientBuilder builder = RestClient.builder(httpHosts); //构造RestClient对象
        RestHighLevelClient restHighLevelClient = new RestHighLevelClient(builder); //将RestClient转为高级的RestHighLevelClient

        return restHighLevelClient;
    }

    /**
     * 获取集群连接
     */
    private static RestHighLevelClient getClient(String clusterName) {
        return clientMap.get(clusterName);
    }


    /**
     * 新增文档
     * @EsIndexInfo： 索引信息(索引名称，集群名称)
     * @EsSourceData： 数据信息(id，元数据)
     */
    public static boolean insertDoc(EsIndexInfo esIndexInfo, EsSourceData esSourceData) {
        IndexRequest indexRequest = new IndexRequest(esIndexInfo.getIndexName()); //根据索引名称创建IndexRequest对象
        indexRequest.source(esSourceData.getData()); //传入新增的map数据
        indexRequest.id(esSourceData.getDocId()); //指定id，如未指定会自动生成
        try {
            getClient(esIndexInfo.getClusterName()).index(indexRequest, COMMON_OPTIONS); //获取集群连接，发送请求
            return true;
        } catch (Exception e) {
            log.error("insertDoc.exception:{}", e.getMessage(), e);
        }
        return false;
    }

    /**
     * 更新文档
     * @EsIndexInfo： 索引信息(索引名称，集群名称)
     * @EsSourceData： 数据信息(id，元数据)
     */
    public static boolean updateDoc(EsIndexInfo esIndexInfo, EsSourceData esSourceData) {
        UpdateRequest updateRequest = new UpdateRequest();
        updateRequest.index(esIndexInfo.getIndexName());
        updateRequest.id(esSourceData.getDocId());
        updateRequest.doc(esSourceData.getData());
        updateRequest.docAsUpsert(true); // 如果文档不存在，则插入新文档
        try {
            getClient(esIndexInfo.getClusterName()).update(updateRequest, COMMON_OPTIONS);
            return true;
        } catch (Exception e) {
            log.error("updateDoc.exception:{}", e.getMessage(), e);
        }
        return false;
    }

    /**
     * 批量更新文档
     * BulkRequest批量处理，其本质就是将多个普通的CRUD请求组合在一起发送。
     */
    public static boolean batchUpdateDoc(EsIndexInfo esIndexInfo, List<EsSourceData> esSourceDataList) {
        try {
            boolean flag = false; //标志位，检验是否有文档能够修改
            BulkRequest bulkRequest = new BulkRequest();
            for(EsSourceData esSourceData : esSourceDataList) {
                //当文档id不为空时，则更新
                if(StringUtils.isNotBlank(esSourceData.getDocId())) {
                    UpdateRequest updateRequest = new UpdateRequest();
                    updateRequest.index(esIndexInfo.getIndexName());
                    updateRequest.id(esSourceData.getDocId());
                    updateRequest.doc(esSourceData.getData());
                    bulkRequest.add(updateRequest);
                    flag = true;
                }
            }

            //只要flag为true，就代表有文档能够修改
            if(flag) {
                BulkResponse bulk = getClient(esIndexInfo.getClusterName()).bulk(bulkRequest, COMMON_OPTIONS);
                //判断是否有失败的
                if(bulk.hasFailures()) {
                    return false;
                }
            }
            return true;
        } catch (Exception e) {
            log.error("batchUpdateDoc.exception:{}", e.getMessage(), e);
        }
        return false;
    }


    /**
     * 删除文档
     * DeleteByQueryRequest：根据条件批量删除文档
     */
    public static boolean delete(EsIndexInfo esIndexInfo) {
        DeleteByQueryRequest deleteByQueryRequest = new DeleteByQueryRequest(esIndexInfo.getIndexName());
        deleteByQueryRequest.setQuery(QueryBuilders.matchAllQuery()); //删除所有
        try {
            BulkByScrollResponse response = getClient(esIndexInfo.getClusterName())
                    .deleteByQuery(deleteByQueryRequest, COMMON_OPTIONS);

            long deleted = response.getDeleted(); //删除的文档数量
            log.info("deleted.size:{}", deleted);
            return true;
        } catch (Exception e) {
            log.error("delete.exception:{}", e.getMessage(), e);
        }
        return false;
    }

    /**
     * 根据id删除文档
     */
    public static boolean deleteDoc(EsIndexInfo esIndexInfo, String docId) {
        DeleteRequest deleteRequest = new DeleteRequest(esIndexInfo.getIndexName());
        deleteRequest.id(docId);
        try {
            DeleteResponse response = getClient(esIndexInfo.getClusterName()).delete(deleteRequest, COMMON_OPTIONS);
            log.info("deleteDoc.size:{}", JSON.toJSONString(response));
            return true;
        } catch (Exception e) {
            log.error("deleteDoc.exception:{}", e.getMessage(), e);
        }
        return false;
    }

    /**
     * 检查文档是否存在
     */
    public static boolean isExistDocById(EsIndexInfo esIndexInfo, String docId) {
        GetRequest getRequest = new GetRequest(esIndexInfo.getIndexName());
        getRequest.id(docId);
        try {
            return getClient(esIndexInfo.getClusterName()).exists(getRequest, COMMON_OPTIONS);
        } catch (IOException e) {
            log.error("isExistDocById.exception:{}", e.getMessage(), e);
        }
        return false;
    }

    /**
     * 根据id查询文档指定字段的值(搜索结果只包含指定的字段)
     *
     * @fields：想要获取的字段name FetchSourceContext 构造函数：
     * fetchSource：是否要从搜索结果中获取文档的源数据, false 时，Elasticsearch 不会返回任何源数据
     * includes：搜索结果中只包含哪些字段，null或空时，全部字段
     * excludes：搜索结果中排除那些字段，null或空时，不排除任何字段
     */
    public static Map<String, Object> getDocById(EsIndexInfo esIndexInfo, String docId, String[] fields) {
        GetRequest getRequest = new GetRequest(esIndexInfo.getIndexName());
        getRequest.id(docId);
        FetchSourceContext fetchSourceContext = new FetchSourceContext(true, fields, null);
        //GetRequest设置fetchSourceContext属性
        getRequest.fetchSourceContext(fetchSourceContext);
        try {
            GetResponse response = getClient(esIndexInfo.getClusterName()).get(getRequest, COMMON_OPTIONS);
            Map<String, Object> source = response.getSource();
            return source;
        } catch (IOException e) {
            log.error("getDocById.exception:{}", e.getMessage(), e);
        }
        return null;
    }

    /**
     * 精确搜索
     * @param esSearchRequest es查询条件
     */
    public static SearchResponse searchWithTermQuery(EsIndexInfo esIndexInfo, EsSearchRequest esSearchRequest) {
        try {
            BoolQueryBuilder bq = esSearchRequest.getBq(); //搜索条件
            String[] fields = esSearchRequest.getFields(); //搜索字段
            int from = esSearchRequest.getFrom(); //分页起始位置
            int size = esSearchRequest.getSize(); //页容量
            Long minutes = esSearchRequest.getMinutes(); //快照缓存时间
            Boolean needScroll = esSearchRequest.getNeedScroll(); //是否需要快照
            String sortName = esSearchRequest.getSortName(); //排序字段
            SortOrder sortOrder = esSearchRequest.getSortOrder(); //排序方式

            //构建搜索条件（fetchSource搜索结果中包含哪些字段信息）
            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
            searchSourceBuilder.query(bq);
            searchSourceBuilder.fetchSource(fields, null).from(from).size(size);

            //设置高亮显示
            if (Objects.nonNull(esSearchRequest.getHighlightBuilder())) {
                searchSourceBuilder.highlighter(esSearchRequest.getHighlightBuilder());
            }

            //设置排序字段
            if (StringUtils.isNotBlank(sortName)) {
                searchSourceBuilder.sort(sortName);
            }

            //设置排序字段
            if(Objects.nonNull(sortOrder)) {
                searchSourceBuilder.sort(new ScoreSortBuilder().order(sortOrder));
            } else {
                searchSourceBuilder.sort(new ScoreSortBuilder().order(SortOrder.DESC));
            }


            //构建搜索请求
            SearchRequest searchRequest = new SearchRequest();
            searchRequest.searchType(SearchType.DEFAULT);
            searchRequest.indices(esIndexInfo.getIndexName());
            searchRequest.source(searchSourceBuilder);
            //设置快照
            if (needScroll) {
                Scroll scroll = new Scroll(TimeValue.timeValueMinutes(minutes));
                searchRequest.scroll(scroll);
            }
            SearchResponse search = getClient(esIndexInfo.getClusterName()).search(searchRequest, COMMON_OPTIONS);
            return search;
        } catch (Exception e) {
            log.error("searchWithTermQuery.exception:{}", e.getMessage(), e);
        }
        return null;
    }
}
